/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.extensions.sql.impl.planner;



import com.bff.gaia.unified.sdk.annotations.Internal;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rel.UnifiedRelNode;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.*;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.ImmutableList;

import org.apache.calcite.plan.RelOptRule;

import org.apache.calcite.rel.RelNode;

import org.apache.calcite.rel.rules.*;

import org.apache.calcite.tools.RuleSet;

import org.apache.calcite.tools.RuleSets;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedAggregationRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedBasicAggregationRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedCalcRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedEnumerableConverterRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedIntersectRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedJoinRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedMinusRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedSortRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedUncollectRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedUnionRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedUnnestRule;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rule.UnifiedValuesRule;



import java.util.List;



/**

 * {@link RuleSet} used in {@code UnifiedQueryPlanner}. It translates a standard Calcite {@link

 * RelNode} tree, to represent with {@link UnifiedRelNode}

 */

@Internal

public class UnifiedRuleSets {

  private static final List<RelOptRule> LOGICAL_OPTIMIZATIONS =

      ImmutableList.of(

          // Rules so we only have to implement Calc

          FilterCalcMergeRule.INSTANCE,

          ProjectCalcMergeRule.INSTANCE,

          FilterToCalcRule.INSTANCE,

          ProjectToCalcRule.INSTANCE,

          // disabled due to https://issues.apache.org/jira/browse/BEAM-6810

          // CalcRemoveRule.INSTANCE,

          CalcMergeRule.INSTANCE,



          // push a filter into a join

          FilterJoinRule.FILTER_ON_JOIN,

          // push filter into the children of a join

          FilterJoinRule.JOIN,

          // push filter through an aggregation

          FilterAggregateTransposeRule.INSTANCE,

          // push filter through set operation

          FilterSetOpTransposeRule.INSTANCE,

          // push project through set operation

          ProjectSetOpTransposeRule.INSTANCE,



          // aggregation and projection rules

          AggregateProjectMergeRule.INSTANCE,

          // push a projection past a filter or vice versa

          ProjectFilterTransposeRule.INSTANCE,

          FilterProjectTransposeRule.INSTANCE,

          // push a projection to the children of a join

          // merge projections

          ProjectMergeRule.INSTANCE,

          // ProjectRemoveRule.INSTANCE,

          // reorder sort and projection

          SortProjectTransposeRule.INSTANCE,

          ProjectSortTransposeRule.INSTANCE,



          // join rules

          JoinPushExpressionsRule.INSTANCE,



          // remove union with only a single child

          UnionEliminatorRule.INSTANCE,

          // convert non-all union into all-union + distinct

          UnionToDistinctRule.INSTANCE,



          // remove aggregation if it does not aggregate and input is already distinct

          AggregateRemoveRule.INSTANCE,

          // push aggregate through join

          AggregateJoinTransposeRule.EXTENDED,

          // aggregate union rule

          AggregateUnionAggregateRule.INSTANCE,



          // reduce aggregate functions like AVG, STDDEV_POP etc.

          // AggregateReduceFunctionsRule.INSTANCE,



          // remove unnecessary sort rule

          // https://issues.apache.org/jira/browse/BEAM-5073

          // SortRemoveRule.INSTANCE,



          // prune empty results rules

          PruneEmptyRules.AGGREGATE_INSTANCE,

          PruneEmptyRules.FILTER_INSTANCE,

          PruneEmptyRules.JOIN_LEFT_INSTANCE,

          PruneEmptyRules.JOIN_RIGHT_INSTANCE,

          PruneEmptyRules.PROJECT_INSTANCE,

          PruneEmptyRules.SORT_INSTANCE,

          PruneEmptyRules.UNION_INSTANCE);



  private static final List<RelOptRule> BEAM_CONVERTERS =

      ImmutableList.of(

          UnifiedCalcRule.INSTANCE,

          UnifiedAggregationRule.INSTANCE,

          UnifiedBasicAggregationRule.INSTANCE,

          UnifiedSortRule.INSTANCE,

          UnifiedValuesRule.INSTANCE,

          UnifiedIntersectRule.INSTANCE,

          UnifiedMinusRule.INSTANCE,

          UnifiedUnionRule.INSTANCE,

          UnifiedUncollectRule.INSTANCE,

          UnifiedUnnestRule.INSTANCE,

          UnifiedJoinRule.INSTANCE);



  private static final List<RelOptRule> BEAM_TO_ENUMERABLE =

      ImmutableList.of(UnifiedEnumerableConverterRule.INSTANCE);



  public static RuleSet[] getRuleSets() {

    return new RuleSet[] {

      RuleSets.ofList(

          ImmutableList.<RelOptRule>builder()

              .addAll(BEAM_CONVERTERS)

              .addAll(BEAM_TO_ENUMERABLE)

              .addAll(LOGICAL_OPTIMIZATIONS)

              .build())

    };

  }

}